
package cn.qtone.test.pinpoint;

import cn.qtone.qtframe.common.utils.DateUtils;
import cn.qtone.qtframe.common.utils.lang.CollectionUtils;
import cn.qtone.qtframe.data.hbase.client.HBaseAdminTemplate;
import cn.qtone.qtframe.data.hbase.client.HbaseOperations2;
import cn.qtone.qtframe.data.hbase.client.HbaseTemplate2;
import cn.qtone.qtframe.data.hbase.client.PooledHTableFactory;
import cn.qtone.qtframe.data.hbase.client.RowMapper;
import cn.qtone.qtframe.data.hbase.client.distributor.RangeOneByteSimpleHash;
import cn.qtone.qtframe.threadpool.CommonThreadPoolManager;
import com.google.common.base.Stopwatch;
import com.google.common.collect.Lists;
import com.navercorp.pinpoint.common.server.bo.SpanBo;
import com.navercorp.pinpoint.common.server.bo.serializer.RowKeyDecoder;
import com.navercorp.pinpoint.common.server.bo.serializer.trace.v2.TraceRowKeyDecoderV2;
import com.navercorp.pinpoint.common.server.bo.serializer.trace.v2.TraceRowKeyEncoderV2;
import com.navercorp.pinpoint.common.util.TransactionId;
import com.navercorp.pinpoint.web.mapper.SpanMapperV2;
import com.navercorp.pinpoint.web.vo.Range;
import com.navercorp.pinpoint.web.vo.scatter.Dot;
import com.sematext.hbase.wd.RowKeyDistributorByHashPrefix;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Created by tanghc on 17/2/20.
 */
public class SearchIndexTestMain {

	// 声明静态配置
	static Configuration conf = null;
	static HbaseOperations2 template = null;
	static HBaseAdminTemplate adminTemplate = null;
	final static ConcurrentHashMap<String,Integer> map = new ConcurrentHashMap<>();
	final static AtomicLong count = new AtomicLong(0);
	static {
		conf = HBaseConfiguration.create();
//		conf.set("hbase.zookeeper.quorum", "localhost");
//		conf.set("hbase.zookeeper.quorum", "172.31.64.11");
		conf.set("hbase.zookeeper.quorum", "10.254.141.132");
		conf.set("hbase.zookeeper.property.clientPort", "2181");
		conf.setLong(HConstants.HBASE_REGIONSERVER_LEASE_PERIOD_KEY, 1200000);
		HbaseTemplate2 hbaseTemplate2 = new HbaseTemplate2();
		hbaseTemplate2.setConfiguration(conf);
		hbaseTemplate2.setTableFactory(new PooledHTableFactory(conf));
		hbaseTemplate2.afterPropertiesSet();
		template = hbaseTemplate2;
		adminTemplate = new HBaseAdminTemplate(conf);
	}
static HbaseOperations2 newHbaseOperations2(){
	HbaseTemplate2 hbaseTemplate2 = new HbaseTemplate2();
	hbaseTemplate2.setConfiguration(conf);
	hbaseTemplate2.setTableFactory(new PooledHTableFactory(conf));
	hbaseTemplate2.afterPropertiesSet();
	return hbaseTemplate2;
}

	public static void main(String[] args)
		throws Exception {
		createIndexScan();
//		trace();
	}

	private static void createIndexScan() {
		final long start = DateUtils.parse("2017-02-20 09:00:00", "yyyy-MM-dd HH:mm:ss").getTime();
		final long end = DateUtils.parse("2017-02-20 18:00:00", "yyyy-MM-dd HH:mm:ss").getTime();
		List<Dot> dots = new ArrayList<>();
		createIndexScan(dots, start, end);
		if(null != dots){
//			for(Dot dot:dots){
//				System.out.println("######"+dot);
//			}
			System.out.println("#####size:"+dots.size());
		}
	}
	private static void trace() throws InterruptedException {
		Stopwatch stopwatch = Stopwatch.createStarted();
		final long start = DateUtils.parse("2017-02-20 09:00:00", "yyyy-MM-dd HH:mm:ss").getTime();
		final long end = DateUtils.parse("2017-02-20 18:00:00", "yyyy-MM-dd HH:mm:ss").getTime();
		List<Dot> dots = new ArrayList<>();
		createIndexScan(dots, start, end);
		if(CollectionUtils.isNotEmpty(dots)){
			System.out.println("#####size:"+dots.size());
		}else{
			System.out.println("#######empty");
			return ;
		}
		final List<TransactionId> transactionIdList = new ArrayList<>();
		CollectionUtils.visitIterator(dots, new CollectionUtils.BeanVisitor<Dot>() {
			@Override
			public void visit(Dot bean) {
				System.out.println("#########"+bean.getTransactionIdAsString());
				transactionIdList.add(bean.getTransactionId());
			}
		});
		int threadSize = 8;
		int sizePerThread = 500;
		final List<List<TransactionId>> partition = Lists.partition(transactionIdList, sizePerThread);

//		List<SpanBo> allSpan = new ArrayList<>();
		if(null != partition){
			final int totolSize = partition.size();
			final int sizePerPage = totolSize / threadSize ;
			final CountDownLatch countDownLatch = new CountDownLatch(threadSize);
			for(  int i =0;i<threadSize;i++){
				final int j = i;
				CommonThreadPoolManager.getInstance().addThread(new Runnable() {
					@Override
					public void run() {
						try {
							List<List<TransactionId>> lists = partition.subList(j * sizePerPage, Math.min((j + 1)*sizePerPage, totolSize));
							if(null != lists){
								for(List<TransactionId> list: lists){
									listTrace(list);
								}
							}
						} finally {
							countDownLatch.countDown();
						}
					}
				});

			}
			CommonThreadPoolManager.getInstance().shutdown();
			countDownLatch.await();
		}

		System.out.println("=======count="+count.longValue()+":times="+stopwatch.elapsed(TimeUnit.SECONDS));
		write(-1, map);

	}
	static List<List<SpanBo>> listTrace(List<TransactionId> transactionIdList){
		final List<Get> multiGet = new ArrayList<>(transactionIdList.size());
		RangeOneByteSimpleHash oneByteSimpleHash = new RangeOneByteSimpleHash(32,40,256);
		RowKeyDistributorByHashPrefix rowKeyDistributorByHashPrefix = new RowKeyDistributorByHashPrefix(oneByteSimpleHash);
		TraceRowKeyEncoderV2 rowKeyEncoder = new TraceRowKeyEncoderV2(rowKeyDistributorByHashPrefix);

		for (TransactionId transactionId : transactionIdList) {
			byte[] transactionIdRowKey = rowKeyEncoder.encodeRowKey(transactionId);
			final Get get = new Get(transactionIdRowKey);
			get.addFamily(com.navercorp.pinpoint.common.hbase.HBaseTables.TRACE_V2_CF_SPAN);
			multiGet.add(get);
		}
		RowKeyDecoder rowKeyDecoder = new TraceRowKeyDecoderV2();
		final SpanMapperV2 spanMapperV2 = new SpanMapperV2(rowKeyDecoder);
		List<List<SpanBo>> lists = template.get(com.navercorp.pinpoint.common.hbase.HBaseTables.TRACE_V2, multiGet, new RowMapper<List<SpanBo>>() {
			@Override
			public List<SpanBo> mapRow(Result result, int i) throws Exception {
				List<SpanBo> spanBos = spanMapperV2.mapRow(result, i);
				if (null != spanBos) {
					for (SpanBo spanBo : spanBos) {
						count(spanBo.getRpc(), map);
						count.incrementAndGet();
					}
				}
				return spanBos;
			}
		});
		return lists;
	}
	private static void createIndexScan(List<Dot> allList,long start,long end) {
		RowKeyDistributorByHashPrefix.OneByteSimpleHash oneByteSimpleHash = new RowKeyDistributorByHashPrefix.OneByteSimpleHash(32);
		RowKeyDistributorByHashPrefix rowKeyDistributorByHashPrefix = new RowKeyDistributorByHashPrefix(oneByteSimpleHash);
		HbaseApplicationTraceIndexDao hbaseApplicationTraceIndexDao = new HbaseApplicationTraceIndexDao();
		hbaseApplicationTraceIndexDao.setHbaseOperations2(template);
		hbaseApplicationTraceIndexDao.setTraceIdRowKeyDistributor(rowKeyDistributorByHashPrefix);
		final Range range = Range.createUncheckedRange(start, end);
		List<Dot> dots = hbaseApplicationTraceIndexDao.scanTraceScatterData("ucenter-pro-uic", range, 5000);
//		List<Dot> dots = hbaseApplic、ationTraceIndexDao.scanTraceScatterData(null, range, 5000);
		long min = Long.MAX_VALUE;
		if(CollectionUtils.isNotEmpty(dots)){
			allList.addAll(dots);
			for(Dot d: dots){
				min = Math.min(min,d.getAcceptedTime());
			}
			System.out.println("start="+new Date(start)+",end="+new Date(end)+",min="+new Date(min));
			if(min <= start){
				return ;
			}
			createIndexScan(allList,start,min);
		}else{
			return ;
		}
	}
	private static void write(int i,Map<String,Integer> map){
		try {
			File file = new File("/tmp/rpc"+i+".txt");
			File fileOther = new File("/tmp/rpc-other"+i+".txt");
			FileWriter fileWriter = new FileWriter(file);
			FileWriter fileOtherWriter = new FileWriter(fileOther);
			for(Map.Entry<String,Integer> entry: map.entrySet()){
				if(entry.getKey().indexOf(":") != -1){
					fileWriter.write(entry.getKey()+"\t"+entry.getValue());
					fileWriter.write("\n");
				}else{
					fileOtherWriter.write(entry.getKey()+"\t"+entry.getValue());
					fileOtherWriter.write("\n");
				}
			}
			fileWriter.flush();
			fileOtherWriter.flush();
		} catch (IOException e) {
			e.printStackTrace();
		}

	}
	private static void count(String key,Map<String,Integer> map){
		if(null == key){
			return ;
		}
		Integer count = map.get(key);
		if(count == null){
			count = 0;
		}
		count ++;
		map.put(key,count);
	}
}
